package com.business.lgh.naocs.naming.consistency.ephemeral.distro;

import com.business.lgh.naocs.naming.consistency.*;
import com.business.lgh.naocs.naming.consistency.ephemeral.EphemeralConsistencyService;
import com.business.lgh.naocs.naming.core.ServerMemberManager;
import com.business.lgh.naocs.naming.core.ServiceManager;
import com.business.lgh.naocs.naming.core.pojo.Instances;
import com.business.lgh.naocs.naming.core.pojo.Member;
import com.business.lgh.naocs.naming.misc.Datum;
import com.business.lgh.naocs.naming.misc.NamingProxy;
import javafx.util.Pair;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author lgh
 * @version V1.0
 * @Package com.business.lgh.naocs.naming.consistency.ephemeral.distro
 * @date 2020/6/24 6:44 下午
 */

@DependsOn("serverMemberManager")
@Service
@Data
public class DistroConsistencyServiceImpl implements EphemeralConsistencyService {

    @Autowired
    DataStore dataStore;

    private Map<String, ConcurrentLinkedQueue<RecordListener>> listeners = new ConcurrentHashMap<>();

    @Autowired
    ServerMemberManager serverMemberManager;

    @Autowired
    ServiceManager serviceManager;

    private volatile Notifier notifier = new Notifier();

    boolean initized = false;

    @PostConstruct
    public void init() {
        //集群数据同步
        GlobalExecutorPools.submitNow(new DataInit());
        //开启线程
        new Thread(notifier).start();
    }

    @Override
    public Datum get(String key) {
        return dataStore.get(key);
    }

    @Override
    public void put(String key, Record value) {
        onPut(key, value);
    }

    @Override
    public void listen(String key, RecordListener listener) {
        if (!listeners.containsKey(key)) {
            listeners.put(key, new ConcurrentLinkedQueue<RecordListener>());
        }
        if (listeners.get(key).contains(listener)) {
            return;
        }
        listeners.get(key).add(listener);
    }

    public void onPut(String key, Record value) {
        Datum datum = new Datum();
        datum.setValue((Instances) value);
        datum.getTimestamp().incrementAndGet();
        //完成datastore的更新
        dataStore.put(key, datum);

        if (!listeners.containsKey(key)) {
            return;
        }

        notifier.addTask(key, ApplyAction.CHANGE);
    }


    public class Notifier implements Runnable {

        private ConcurrentHashMap<String, String> services = new ConcurrentHashMap<>();

        private BlockingQueue<Pair<String, ApplyAction>> tasks = new ArrayBlockingQueue<>(1024 * 1024);

        public void addTask(String datumKey, ApplyAction action) {
            //如果service已包含这个key，就代表已有了
            if (services.contains(datumKey) && action == ApplyAction.CHANGE) {
                return;
            }
            if (action == ApplyAction.CHANGE) {
                services.put(datumKey, StringUtils.EMPTY);
            }
            tasks.offer(new Pair<>(datumKey, action));
        }

        @Override
        public void run() {
            for (; ; ) {
                try {
                    Pair<String, ApplyAction> take = tasks.take();
                    //处理pair
                    handle(take);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        private void handle(Pair<String, ApplyAction> pair) {
            String datumkey = pair.getKey();
            ApplyAction action = pair.getValue();
            //代表该任务开始消费了
            services.remove(datumkey);

            //如果没服务监听则直接返回
            if (!listeners.containsKey(datumkey)) {
                return;
            }

            //遍历监听,主要是把datastore的key和datum里的集合通知service进行数据更新
            for (RecordListener listener : listeners.get(datumkey)) {
                if (action == ApplyAction.CHANGE) {
                    listener.onChange(datumkey, dataStore.get(datumkey).value);
                }
            }
        }
    }


    public class DataInit implements Runnable {

        @Override
        public void run() {
            try {
                load();
                if (!initized) {
                    GlobalExecutorPools.submitDelay(this,30);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        }

        private void load() throws InterruptedException {
            Map<String, Member> serverList = serverMemberManager.getServerList();

            //若server列表为空，则循环等待
            while (serverList.size() < 1) {
                Thread.sleep(1000);
            }

            //遍历集合循环
            for (Map.Entry<String, Member> entries : serverList.entrySet()) {
                Member member = entries.getValue();
                //匹配到是自己IP的则忽略
                if (entries.getKey().equals(serverMemberManager.getLocalAddress())) {
                    continue;
                }
                if (syncAllDataFromRemote(entries.getKey())) {
                    initized = true;
                    return;
                }
            }
        }

        //从其他server同步数据
        private boolean syncAllDataFromRemote(String serverAdress) {
            try {
                DataStore remoteData = NamingProxy.syncGetDataStore(serverAdress);
                if (remoteData == null) {
                    return false;
                }
                //遍历同步数据
                remoteData.getDatasource().forEach((k, v) -> {
                    if (dataStore.get(k) == null) {
                        ((Instances)((Datum)v).getValue()).getInstances().forEach(i->{
                            serviceManager.registerInstance(i.getNamespaceId(),i.getServiceName(),i);
                        });
                    }
                });
            } catch (IOException e) {
                e.printStackTrace();
                return false;
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
                return false;
            }
            return true;
        }

    }

}
